package com.atbeijing.D08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * 另一种方式写连续三次登陆失败数据筛选
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //数据源
        SingleOutputStreamOperator<Event> stream = env
                .fromElements(
                        new Event("user-2", "fail", 1000L),
                        new Event("user-1", "success", 2000L),
                        new Event("user-1", "fail", 3000L),
                        new Event("user-1", "fail", 4000L),
                        new Event("user-1", "fail", 5000L),
                        new Event("user-1", "success", 6000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        //创建模板
        Pattern<Event, Event> pattern = Pattern
                .<Event>begin("fail")//连续三个事件的名
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event value) throws Exception {
                        return value.eventType.equals("fail");
                    }
                })
                .times(3);//连续三个事件符合where的要求

        //通过模板处理数据
        PatternStream<Event> patternStream = CEP.pattern(stream.keyBy(r -> r.userId), pattern);

        //处理符合要求的数据
        SingleOutputStreamOperator<String> result = patternStream.select(new PatternSelectFunction<Event, String>() {
            @Override
            public String select(Map<String, List<Event>> pattern) throws Exception {
                List<Event> fail = pattern.get("fail");
                Event first = fail.get(0);
                Event second = fail.get(1);
                Event third = fail.get(2);
                return "用户" + first.userId + "在" + first.timestamp + "," + second.timestamp + "," + third.timestamp + "时间登录失败！";
            }
        });

        result.print();

        env.execute();
    }

    public static class Event{
        public String userId;
        public String eventType;
        public Long timestamp;

        public Event() {
        }

        public Event(String userId, String eventType, Long timestamp) {
            this.userId = userId;
            this.eventType = eventType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "userId='" + userId + '\'' +
                    ", eventType='" + eventType + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}
